package cn.zynworld.hnister.chats.manager;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * @auther Buynow Zhao
 * @create 2018/5/6
 *
 */
@Component
public class MqttManager {

	private Logger logger = LoggerFactory.getLogger(MqttClient.class);
	private ObjectMapper mapper = new ObjectMapper();

	@Value("${mqtt.broker}")
	private String MQTT_BROKER;
	@Value("${mqtt.qos}")
	private Integer MQTT_QOS;
	@Value("${mqtt.clientId}")
	private String MQTT_CLIENTID;
	@Value("${mqtt.username}")
	private String MQTT_USERNAME;
	@Value("${mqtt.password}")
	private String MQTT_PASSWORD;

	private MqttClient mqttClient;


	@PostConstruct
	public void init() throws MqttException{
		// init the mqtt component
		mqttClient = new MqttClient(MQTT_BROKER, MQTT_CLIENTID, new MemoryPersistence());
		MqttConnectOptions connOpts = new MqttConnectOptions();
		connOpts.setCleanSession(false);
		connOpts.setUserName(MQTT_USERNAME);
		connOpts.setPassword(MQTT_PASSWORD.toCharArray());
		mqttClient.connect(connOpts);
		logger.info("建立与mqtt server连接");
	}

	//保证与mqtt的连通
	public void keepConnect() {
		try {
			if (!mqttClient.isConnected()) {
				mqttClient.reconnect();
			}
		} catch (MqttException e) {
			e.printStackTrace();
		}

	}

	@PreDestroy
	public void destory() throws MqttException {
		logger.info("与mqtt连接将被取消");
		mqttClient.disconnect();
		logger.info("与mqtt连接取消成功");
	}

	public void publish(String topic, String msg, int qos, boolean retained) {
		try {
			keepConnect();
			mqttClient.publish(topic,msg.getBytes(),qos,retained);
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}

	public void publish(String topic, Object msg, int qos, boolean retained) {
		try {
			publish(topic,mapper.writeValueAsString(msg),qos,retained);
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
	}

	public void publish(String topic, MqttMessage msg) {
		try {
			keepConnect();
			mqttClient.publish(topic,msg);
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}

	public void subscribe(String topicFilter, IMqttMessageListener listener){
		try {
			mqttClient.subscribe(topicFilter, listener);
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}






}
